package aldwxstat

import java.sql.{Connection, DriverManager}

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.functions.countDistinct
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import aldwxutils.{AldwxDebug, DBConf, TimeUtil}
import org.apache.spark.storage.StorageLevel

/**
  * Created by ald on 2017/08/02.
  * 二维码扫码个数,扫码人数,扫码次数,扫码新增
  */
object QrcodeHourlyAndDaily {
  def main(args: Array[String]) {
    Logger.getLogger("org").setLevel(Level.WARN)

    // 打印调试信息
    AldwxDebug.debug_info("2017-11-01", "sunxiaowei")

    val spark = SparkSession
      .builder()
      .appName(this.getClass.getName)
        .master("local[*]")
      //.config("spark.sql.shuffle.partitions", 240)
      .getOrCreate()

    // 获取 ald_code 的信息
    val code_df = DBConf.read_from_mysql(spark, "(select app_key,qr_key from ald_code) as code_df")

    // 获取指定日期
    val the_day = TimeUtil.processArgs(args)
    println(the_day)

    //val df:DataFrame = spark.read.parquet(s"${DBConf.hdfsUrl}/${the_day}/*/part-*").repartition(120).cache()
    val df:DataFrame = spark.read.parquet("d:/test/parquet/20171106/*/part-*").repartition(120)
    qrcodeDaily(the_day, df)
    qrcodeHourly(the_day, df)
    spark.stop()
  }

  def qrcodeDaily(day:String, df:DataFrame): Unit = {

    // 判断某天是否上报二维码数据
    if (!df.columns.contains("qr")) return

    // 二维码数据源
    val page_df = df.filter("ev='page' and ag_aldsrc !='null' and qr !='null'")
      .select(
        df("ak").alias("app_key"),
        df("qr").alias("qr_key"),
        df("at"),
        df("uu")
      ).cache()

    //page_df.show()

    // 扫码人数,扫码次数
    val qr_count_df = page_df.groupBy("app_key", "qr_key")
      .agg(
        countDistinct("uu") as "total_scan_user_count",
        countDistinct("at") as "total_scan_count"
      )

    // 扫码新增
    val app_df = df.filter("ev='app' and ifo='true'")
      .select(
        df("ak").alias("app_key"),
        df("at"),
        df("uu")
      )



    val qr_newer_df = page_df.join(
      app_df,
        page_df("app_key")===app_df("app_key") &&
        page_df("at")===app_df("at") &&
        page_df("uu")===app_df("uu")
    ).select(
      page_df("app_key"),
      page_df("qr_key"),
      page_df("uu")
    ).groupBy("app_key", "qr_key")
      .agg(
        countDistinct("uu").alias("qr_new_comer_for_app")
      )

    // qr_newer_df.show()

    val _result_df = qr_count_df.join(
      qr_newer_df,
      qr_count_df("app_key")===qr_newer_df("app_key") &&
        qr_count_df("qr_key")===qr_newer_df("qr_key"),
      "leftouter"
    ).select(
      qr_count_df("app_key"),
      qr_count_df("qr_key"),
      qr_count_df("total_scan_user_count"),
      qr_count_df("total_scan_count"),
      qr_newer_df("qr_new_comer_for_app")
    ).na.fill(0)

    _result_df.show(5)
    // 二维码每日统计
    //qrcodeDailyForeachPartition(day, _result_df)
  }

  def qrcodeHourly(day:String, df:DataFrame): Unit = {
    // 二维码数据源
    if (!df.columns.contains("qr")) return
    val page_df = df.filter("ev='page' and ag_aldsrc !='null' and qr !='null' and ifp='true'")
      .select(
        df("ak").alias("app_key"),
        df("hour"),
        df("qr").alias("qr_key"),
        df("at"),
        df("uu")
      ).cache()

    // 扫码人数,扫码次数
    val qr_count_df = page_df.groupBy("app_key", "qr_key", "hour")
      .agg(
        countDistinct("uu") as "qr_visitor_count",
        countDistinct("at") as "qr_scan_count"
      )

    // 新增人数
    val app_df = df.filter("ev='app' and ifo='true'")
      .select(
        df("ak").alias("app_key"),
        df("hour"),
        df("uu"),
        df("at")
      )

    // 扫码新增人数
    val qr_newer_df = page_df.join(
      app_df,
      page_df("app_key")===app_df("app_key") &&
        page_df("at")===app_df("at") &&
        page_df("hour")===app_df("hour") &&
        page_df("uu")===app_df("uu")
    ).select(
      page_df("app_key"),
      page_df("hour"),
      page_df("qr_key"),
      app_df("uu")
    ).groupBy("app_key", "qr_key", "hour")
      .agg(
        countDistinct("uu").alias("qr_newer_count")
      )

    //qr_newer_df.show()

    // 生成结果表
    val _result_df = qr_count_df.join(
      qr_newer_df,
      qr_count_df("app_key")===qr_newer_df("app_key") &&
        qr_count_df("hour")===qr_newer_df("hour") &&
        qr_count_df("qr_key")===qr_newer_df("qr_key"),
      "leftouter"
    ).select(
      qr_count_df("app_key"),
      qr_count_df("hour"),
      qr_count_df("qr_key"),
      qr_count_df("qr_visitor_count"),
      qr_count_df("qr_scan_count"),
      qr_newer_df("qr_newer_count")
    ).na.fill(0)

    _result_df.show(5)
    // 二维码每小时统计
   // qrcodeHourlyForeachPartition(day, _result_df)
  }
  // 二维码每日入库
  private def qrcodeDailyForeachPartition(day:String,qrcodeDailyDf:DataFrame) = {
    // 创建 mysql connect
    val url         = DBConf.url
    val username    = DBConf.user
    val password    = DBConf.password
    var connection:Connection = null
    val driver = "com.mysql.jdbc.Driver"
    val update_at = TimeUtil.nowInt()
    val the_day = day

    // 逐行入库
    qrcodeDailyDf.foreachPartition((rows:Iterator[Row]) => {

      try {
        Class.forName(driver)
        connection = DriverManager.getConnection(url,username,password)
        connection.setAutoCommit(false)
        val statement = connection.createStatement

        rows.foreach(r=>{
          val (app_key,qr_key,total_scan_user_count,total_scan_count,qr_new_comer_for_app) = ( r(0),r(1),r(2),r(3),r(4) )

          val sql = s"insert into aldstat_qr_code_statistics (app_key,day,qr_key,total_scan_user_count," +
            s"total_scan_count,qr_new_comer_for_app, update_at)" +
            s"values ('${app_key}', '${the_day}', '${qr_key}', '${total_scan_user_count}', '${total_scan_count}', '${qr_new_comer_for_app}','${update_at}') ON DUPLICATE KEY UPDATE total_scan_user_count='${total_scan_user_count}', total_scan_count='${total_scan_count}', qr_new_comer_for_app='${qr_new_comer_for_app}',update_at='${update_at}'"
          statement.addBatch(sql)
        })
        statement.executeBatch
        connection.commit()
      } catch {
        case e: Exception => e.printStackTrace()
          connection.close()
      }
    })
  }

  private def qrcodeHourlyForeachPartition(day:String,qrcodeHourlyDf:DataFrame) = {
    // 创建 mysql connect
    val url         = DBConf.url
    val username    = DBConf.user
    val password    = DBConf.password
    var connection:Connection = null
    val driver = "com.mysql.jdbc.Driver"
    val update_at = TimeUtil.ytoday()
    val the_day = day

    // 逐行入库
    qrcodeHourlyDf.foreachPartition((rows:Iterator[Row]) => {

      try {
        Class.forName(driver)
        connection = DriverManager.getConnection(url,username,password)
        connection.setAutoCommit(false)
        val statement = connection.createStatement

        rows.foreach(r=>{
          val app_key=r(0)
          val hour=r(1)
          val qr_key=r(2)
          val qr_visitor_count=r(3)
          val qr_scan_count=r(4)
          val qr_newer_count=r(5)

          val sql = s"insert into aldstat_hourly_qr (app_key,day,hour,qr_key,qr_visitor_count," +
            s"qr_scan_count,qr_newer_count, update_at)" +
            s"values ('${app_key}', '${the_day}', '${hour}', '${qr_key}', '${qr_visitor_count}', '${qr_scan_count}','${qr_newer_count}', '${update_at}') ON DUPLICATE KEY UPDATE qr_visitor_count='${qr_visitor_count}', qr_scan_count='${qr_scan_count}', qr_newer_count='${qr_newer_count}',update_at='${update_at}'"
          statement.addBatch(sql)
        })
        statement.executeBatch
        connection.commit()
      } catch {
        case e: Exception => e.printStackTrace()
          connection.close()
      }
    })
  }
}